package com.xzx.ecommercedw.flume.interceptor;

import lombok.extern.slf4j.Slf4j;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 校验数据是否合法，不合法直接丢弃
 *
 * @author xinzhixuan
 * @version 1.0
 * @date 2024/9/17 18:12
 */
@Slf4j
public class LogETLInterceptor implements Interceptor {
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        byte[] body = event.getBody();
        String logInfo = new String(body, StandardCharsets.UTF_8);
        if (logInfo.contains("start")) {
            if (validateStart(logInfo)) {
                return event;
            }
        } else {
            if (validateEvent(logInfo)) {
                return event;
            }
        }
        // 返回null表示丢弃数据
        log.info("invalid data:{}", logInfo);
        return null;
    }

    /**
     * // 服务器时间 | json
     * // 1549696569054 | {"cm":{"ln":"-89.2","sv":"V2.0.4","os":"8.2.0","g":"M67B4QYU@gmail.com","nw":"4G","l":"en","vc":"18","hw":"1080*1920","ar":"MX","uid":"u8678","t":"1549679122062","la":"-27.4","md":"sumsung-12","vn":"1.1.3","ba":"Sumsung","sr":"Y"},"ap":"weather","et":[]}
     *
     * @param logInfo .
     * @return .
     */
    private boolean validateEvent(String logInfo) {
        return true;
    }

    /**
     * // {"action":"1","ar":"MX","ba":"HTC","detail":"542","en":"start","entry":"2","extend1":"","g":"S3HQ7LKM@gmail.com","hw":"640*960","l":"en","la":"-43.4","ln":"-98.3","loading_time":"10","md":"HTC-5","mid":"993","nw":"WIFI","open_ad_type":"1","os":"8.2.1","sr":"D","sv":"V2.9.0","t":"1559551922019","uid":"993","vc":"0","vn":"1.1.5"}
     *
     * @param logInfo .
     * @return .
     */
    private boolean validateStart(String logInfo) {
        return true;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        return events.stream().map(this::intercept).filter(Objects::nonNull).collect(Collectors.toList());
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new LogETLInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}
